package com.monkeyboy.demo.adminapi;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @Author Gavin
 * @date 2020.08.23 16:46
 */
public class AdminSimple {
    public final static String TOPIC_NAME = "test_topic";

    public static void main(String[] args) throws Exception {
//        AdminClient adminClient = getAdminClient();
//        System.out.println(adminClient);
        //创建topic
//        createTopic();
//        topicList();
        //删除topic
//        topicDelete();
//        topicList();
        //查看topic的描述信息
        // topicDescribe();
        //查看config
        //describeConfig();
        //修改配置
//        alterConfig();
//        describeConfig();
        //增加partition
//        incrPartitions(2);
        topicDescribe();
    }

    /**
     * 增加Partition数量
     *
     * @throws Exception
     */
    public static void incrPartitions(int partitions) throws Exception {
        AdminClient adminClient = getAdminClient();
        Map<String, NewPartitions> partitionsMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(partitions);
        partitionsMap.put(TOPIC_NAME, newPartitions);
        CreatePartitionsResult partitions1 = adminClient.createPartitions(partitionsMap);
        partitions1.all().get();
    }

    /**
     * 修改配置
     *
     * @throws Exception
     */
    public static void alterConfig() throws Exception {
        AdminClient adminClient = getAdminClient();

        //old api
//        Map<ConfigResource, Config> configMap = new HashMap<>();
//        //组织两个参数
//        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
//        Config config = new Config(Arrays.asList(new ConfigEntry("preallocate", "true")));
//        configMap.put(configResource, config);
//
//        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configMap);

        //new api
        Map<ConfigResource, Collection<AlterConfigOp>> configMap = new HashMap<>();
        //组织两个参数
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        AlterConfigOp alterConfigOp = new AlterConfigOp(new ConfigEntry("preallocate", "true"), AlterConfigOp.OpType.SET);
        configMap.put(configResource, Arrays.asList(alterConfigOp));

        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(configMap);

        alterConfigsResult.all().get();
    }

    /*
    查看配置
     */
    public static void describeConfig() throws Exception {
        AdminClient adminClient = getAdminClient();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Collections.singletonList(configResource));
        Map<ConfigResource, Config> configResourceConfigMap = describeConfigsResult.all().get();
        configResourceConfigMap.forEach((key, value) -> System.out.println("configResource: " + key + ",config:" + value));
    }

    /*
    topic描述:
    name: test_topic,
    desc:  (
    name=test_topic,
    internal=false,
    partitions=(
    partition=0, leader=192.168.1.150:9092 (id: 0 rack: null),
     replicas=192.168.1.150:9092 (id: 0 rack: null),
     isr=192.168.1.150:9092 (id: 0 rack: null)),
      authorizedOperations=[])
     */
    public static void topicDescribe() throws Exception {
        AdminClient adminClient = getAdminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(TOPIC_NAME));
        Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
        Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
        entries.forEach((pre) -> {
            System.out.println("name: " + pre.getKey() + ",desc:  " + pre.getValue());
        });
    }

    /**
     * 删除topic
     *
     * @throws Exception
     */
    public static void topicDelete() throws Exception {
        AdminClient adminClient = getAdminClient();
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Collections.singletonList(TOPIC_NAME));
        deleteTopicsResult.all().get();//阻塞获取结果
    }

    //获取topic的列表
    public static void topicList() throws Exception {
        AdminClient adminClient = getAdminClient();
        //可以添加额外的一些参数
        ListTopicsOptions options = new ListTopicsOptions();
        options.listInternal(true);//表示是否查看internal选项
        ListTopicsResult listTopicsResult = adminClient.listTopics(options);
        Set<String> names = listTopicsResult.names().get();
        names.forEach(System.out::println);//打印topic的名字
        Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        topicListings.forEach(System.out::println);//可以获取该topic不是属于internal的topic
    }

    /**
     * 创建topic
     *
     * @Param
     * @Date 2020.08.23 17:27
     **/
    public static void createTopic() {
        AdminClient adminClient = getAdminClient();
        short rep = 1;//副本因子，为了保证创建的partition有几个副本，已达到自动备份和恢复数据的作用
        NewTopic newTopic = new NewTopic(TOPIC_NAME, 1, rep);
//        CreateTopicsResult result = adminClient.createTopics(Arrays.asList(newTopic));
        CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
        System.out.println(result.values());
    }

    public static AdminClient getAdminClient() {
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.150:9092");
        AdminClient adminClient = AdminClient.create(properties);
        return adminClient;
    }
}
